package com.pine.dcts.core.event;

import com.alibaba.datax.core.statistics.communication.CommunicationTool;
import com.pine.dcts.entity.JobLog;
import com.pine.dcts.service.JobLogService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;

import java.text.SimpleDateFormat;

@Slf4j
public class JobMessageListener implements MessageListener, ApplicationContextAware {
    private static ApplicationContext applicationContext;
    private  static JobLogService jobLogService ;
    @Override
    public void transmitMessage(JobMessage jobMessage) {
       final SimpleDateFormat dateFormat = new SimpleDateFormat(
                "yyyy-MM-dd HH:mm:ss");
        System.out.println("jobid："+jobMessage.getJobId());
        long totalCosts = (jobMessage.getEndTimeStamp() - jobMessage.getStartTimeStamp()) / 1000;
        long transferCosts = (jobMessage.getEndTransferTimeStamp() - jobMessage.getStartTransferTimeStamp()) / 1000;
        if (0L == transferCosts) {
            transferCosts = 1L;
        }
        // 字节速率
        long byteSpeedPerSecond = jobMessage.getCommunication().getLongCounter(CommunicationTool.READ_SUCCEED_BYTES)
                / transferCosts;

        long recordSpeedPerSecond = jobMessage.getCommunication().getLongCounter(CommunicationTool.READ_SUCCEED_RECORDS)
                / transferCosts;

//        log.info("处理任务："+String.format(
//                "\n" + "%-26s: %-18s\n"+ "%-26s: %-18s\n" + "%-26s: %-18s\n" + "%-26s: %19s\n"
//                        + "%-26s: %19s\n" + "%-26s: %19s\n" + "%-26s: %19s\n"
//                        + "%-26s: %19s\n",
//                "任务id",
//                jobMessage.getJobId(),
//                "任务启动时刻",
//                dateFormat.format(jobMessage.getStartTimeStamp()),
//
//                "任务结束时刻",
//                dateFormat.format(jobMessage.getEndTimeStamp()),
//
//                "任务总计耗时",
//                totalCosts + "s",
//                "任务平均流量",
//                StrUtil.stringify(byteSpeedPerSecond)
//                        + "/s",
//                "记录写入速度",
//               recordSpeedPerSecond
//                        + "rec/s", "读出记录总数",
//               CommunicationTool.getTotalReadRecords(jobMessage.getCommunication()),
//                "读写失败总数",
//               CommunicationTool.getTotalErrorRecords(jobMessage.getCommunication())));

        JobLog jobLog = new JobLog();
        jobLog.setByteSpeedPerSecond(byteSpeedPerSecond);
        jobLog.setJobId(jobMessage.getJobId());
        jobLog.setEndTimeStamp(jobMessage.getEndTimeStamp());
        jobLog.setRecordSpeedPerSecond(recordSpeedPerSecond);
        jobLog.setStartTimeStamp(jobMessage.getStartTimeStamp());
        jobLog.setTotalCosts(totalCosts);
        jobLog.setTotalReadRecords(CommunicationTool.
                getTotalReadRecords(jobMessage.getCommunication()));
        jobLog.setTotalErrorRecords(CommunicationTool.
                getTotalErrorRecords(jobMessage.getCommunication()));
        jobLog.setStatus(1);
        if(jobMessage.getCommunication().getThrowable()!=null){
            jobLog.setStatus(0);
            jobLog.setErrorMessage(jobMessage.getCommunication().getThrowableMessage());
        }
      //  jobLogService.save(jobLog);
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
        jobLogService  = applicationContext.getBean(JobLogService.class);
    }
}
